package com.haiyou.common.thread;

import java.time.Duration;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.commons.lang3.Validate;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.collect.Maps;

import lombok.Getter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;

/**
 * 
 * 
 * @Description: 原子任务执行器
 * @author xingyuan
 * @date 2022年2月15日 下午5:31:22
 */
@Slf4j
public class AtomicExecutor {

	/**
	 * 基于链表的可扩展、高性能、公平的自旋锁
	 * 
	 * @author Administrator
	 *
	 */
	public static final class MCSLock {
		private AtomicReference<QNode> tail;
		private ThreadLocal<QNode> myNode;

		public MCSLock() {
			tail = new AtomicReference<>(null);
			myNode = ThreadLocal.withInitial(QNode::new);
		}

		public void lock() {
			QNode qnode = myNode.get();
			QNode preNode = tail.getAndSet(qnode);
			if (preNode != null) {
				qnode.locked = false;
				preNode.next = qnode;
				// wait until predecessor gives up the lock
				while (!qnode.locked) {
				}
			}
			qnode.locked = true;
		}

		public void unlock() {
			QNode qnode = myNode.get();
			if (qnode.next == null) {
				// 后面没有等待线程的情况
				if (tail.compareAndSet(qnode, null)) {
					// 真的没有等待线程，则直接返回，不需要通知
					return;
				}
				// wait until predecessor fills in its next field
				// 突然有人排在自己后面了，可能还不知道是谁，下面是等待后续者
				while (qnode.next == null) {
				}
			}
			// 后面有等待线程，则通知后面的线程
			qnode.next.locked = true;
			qnode.next = null;
		}

		private class QNode {
			/**
			 * 是否被qNode所属线程锁定
			 */
			private volatile boolean locked = false;
			/**
			 * 与CLHLock相比，多了这个真正的next
			 */
			private volatile QNode next = null;
		}

	}

	@ToString(callSuper = true)
	@Getter
	public static final class LockException extends RuntimeException {

		/**
		 * 
		 */
		private static final long serialVersionUID = -7294539525095965738L;

		public LockException(Object key, Throwable cause) {
			super(String.valueOf(key), cause);
		}

	}

	private static ConcurrentMap<Thread, ConcurrentMap<Object, LockException>> lockMap = Maps.newConcurrentMap();

	private static Cache<Object, MCSLock> cachedLocks = Caffeine.newBuilder().expireAfterAccess(Duration.ofSeconds(60))
			.build();

	private static MCSLock getLock(Object key) {
		return cachedLocks.get(key, k -> new MCSLock());
	}

	public static void run(Object key, Runnable command) {
		call(key, () -> {
			command.run();
			return null;
		});
	}

	public static <V> V call(Object key, Callable<V> command) {
		MCSLock lock = getLock(key);
		Thread thread = Thread.currentThread();
		if (lockMap.containsKey(thread)) {
			ConcurrentMap<Object, LockException> lockExceptionMap = lockMap.get(thread);
			if (lockExceptionMap.containsKey(key)) {
				LockException lockException = lockExceptionMap.get(key);
				if (lockException.getMessage().equals(key)) {
					throw new RuntimeException("嵌套调用锁:" + lockException);
				}
			}
		}
		try {
			lock.lock();
			Validate.isTrue(lockMap.computeIfAbsent(thread, k -> Maps.newConcurrentMap()).putIfAbsent(key,
					new LockException(key, new Throwable())) == null);
			try {
				log.debug("locked [{}]", key);
				return command.call();
			} catch (Exception e) {
				log.error(e.getMessage() + ":" + key, e);
				return null;
			} finally {
				lock.unlock();
				if (lockMap.containsKey(thread)) {
					ConcurrentMap<Object, LockException> lockExceptionMap = lockMap.get(thread);
					if (lockExceptionMap.containsKey(key)) {
						lockExceptionMap.remove(key);
						if (lockExceptionMap.isEmpty()) {
							lockMap.remove(thread);
						}
					}
				}
			}
		} catch (Exception e) {
			log.error(e.getMessage() + ":" + key, e);
			return null;
		}
	}

}
